// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <cantor/catalog/namespace_manager.h>

namespace cantor::catalog {

    turbo::Status NamespaceManager::create_namespace(std::shared_ptr<NamespaceInfo> info, CatalogStorage *storage) {
        std::unique_lock lk(_mutex);

        auto it = _namespace_id_map.find(info->namespace_name());
        if (it != _namespace_id_map.end()) {
            return turbo::already_exists_error("namespace ", info->namespace_name(), "already exists with id:", it->second);
        }

        /// save to storage
        int64_t temp_id = _max_namespace_id + 1;
        info->set_namespace_id(temp_id);
        info->upgrade_version();
        auto key = storage->make_namespace_key(info->namespace_name(), temp_id);
        TURBO_MOVE_OR_RAISE(auto value, info->serialize());
        TURBO_RETURN_NOT_OK(storage->save_namespace(key, value, temp_id));
        /// update memory
        _max_namespace_id = temp_id;
        _namespace_id_map[info->namespace_name()] = temp_id;
        _namespace_map[temp_id] = std::move(info);
        return turbo::OkStatus();
    }

    turbo::Status NamespaceManager::remove_namespace(const std::string &ns_name, CatalogStorage *storage) {
        std::unique_lock lk(_mutex);

        auto it = _namespace_id_map.find(ns_name);
        if (it == _namespace_id_map.end()) {
            return turbo::not_found_error("not found namespace ", ns_name);
        }
        auto id = it->second;
        auto dbs_it = _database_ids.find(id);
        if (dbs_it != _database_ids.end() && !dbs_it->second.empty()) {
            return turbo::failed_precondition_error("namespace: ", ns_name, "already exists databases, must remove all the databases before remove namespace");
        }

        auto key = storage->make_namespace_key(ns_name, id);
        TURBO_RETURN_NOT_OK(storage->remove_namespace(key));
        /// update memory
        _namespace_map.erase(id);
        _database_ids.erase(id);
        _namespace_id_map.erase(ns_name);
        return turbo::OkStatus();
    }

    turbo::Status NamespaceManager::update_namespace(std::shared_ptr<NamespaceInfo> info, CatalogStorage *storage) {
        std::unique_lock lk(_mutex);

        auto it = _namespace_id_map.find(info->namespace_name());
        if (it == _namespace_id_map.end()) {
            return turbo::not_found_error("not found namespace ", info->namespace_name());
        }

        auto id = it->second;
        auto &old = _namespace_map[id];
        TURBO_ABORT_NOT_OK(info->merge_for_update(old.get()));
        info->upgrade_version();
        auto key = storage->make_namespace_key(info->namespace_name(), id);
        TURBO_MOVE_OR_RAISE(auto value , info->serialize());
        TURBO_RETURN_NOT_OK(storage->update_namespace(key, value));

        /// update memory
        _namespace_map[id] = std::move(info);
        return turbo::OkStatus();
    }

    void NamespaceManager::set_max_id(const int64_t max_id) {
        std::unique_lock lk(_mutex);
        _max_namespace_id = max_id;
    }

    turbo::Status NamespaceManager::add_database_id(int64_t ns_id, int64_t db_id) {
        std::unique_lock lk(_mutex);
        auto it = _namespace_map.find(ns_id);
        if (it == _namespace_map.end()) {
            return turbo::not_found_error("not found namespace ", ns_id);
        }
        _database_ids[ns_id].insert(db_id);
        return turbo::OkStatus();
    }

    turbo::Status NamespaceManager::remove_database_id(int64_t ns_id, int64_t db_id) {
        std::unique_lock lk(_mutex);
        auto it = _database_ids.find(ns_id);
        if (it == _database_ids.end()) {
            return turbo::not_found_error("not found namespace ", ns_id);
        }
        it->second.erase(db_id);
        return turbo::OkStatus();
    }

    void NamespaceManager::clear_database(int64_t ns_id) {
        std::unique_lock lk(_mutex);
        _database_ids.erase(ns_id);
    }
}  // namespace cantor::catalog
